/*
 * Copyright 2014 Semidata. All rights reserved.
 */
package com.semidata.rtd.engine.executor;

import java.lang.Thread.UncaughtExceptionHandler;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import com.semidata.rtd.core.service.Service;

/**
 * @author shewei@semidata.com, She Wei
 * 
 */
public class TaskWorker implements Service {

	public enum Status {
		INITIALIZING, IDEL, RUNNING, CANCELLING, DONE, DEAD,
	}

	private static final long KILL_TIME = 5L;

	private String name;
	private Status status;
	private Thread thread = null;
	private boolean quit = false;
	private ReentrantLock lock;
	private Condition cond;
	private TaskWrapper wrapper = null;
	private long cancelTime = 0;

	public TaskWorker(String name) {
		this.name = name;
		this.status = Status.INITIALIZING;
		this.lock = new ReentrantLock();
		this.cond = lock.newCondition();
	}

	@Override
	public void start() {
		thread = new Thread(new Runnable() {

			@Override
			public void run() {
				execute();
			}
		}, name);

		thread.start();
	}

	@Override
	public void stop() {
		quit = true;

		lock.lock();
		try {
			cond.signal();
		} finally {
			lock.unlock();
		}

		if (thread != null) {
			try {
				thread.join();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}

	}

	public Status getStatus() {
		return status;
	}

	private void execute() {
		status = Status.IDEL;
		while (true) {

			try {
				lock.lock();
				try {
					while (wrapper == null && status != Status.RUNNING && !quit) {
						try {
							cond.await();
						} catch (InterruptedException e) {
						}
					}
				} finally {
					lock.unlock();
				}

				if (quit)
					break;

				if (wrapper == null)
					continue;

				Exception exception = null;
				try {
					wrapper.run();
				} catch (Exception e) {
					exception = e;
				}

				lock.lock();
				try {
					if (status != Status.DEAD)
						status = Status.DONE;

				} finally {
					lock.unlock();
				}

				if (status == Status.DONE) {
					wrapper.done(exception);
					lock.lock();
					try {
						wrapper = null;
						status = Status.IDEL;
					} finally {
						lock.unlock();
					}
				}
			} catch (IllegalMonitorStateException e) {
				if (status != Status.DEAD) {
					throw e;
				}
			}

		}

		if (wrapper != null)
			wrapper.done(new TaskExecutorException());
		wrapper = null;
	}

	public void run(TaskWrapper wrapper) {
		lock.lock();
		try {
			assert (this.status == Status.IDEL);
			assert (this.wrapper == null);

			this.status = Status.RUNNING;
			this.wrapper = wrapper;
			cond.signal();
		} finally {
			lock.unlock();
		}
	}

	public void check() {

		lock.lock();
		try {
			if (wrapper == null)
				return;
			long now = System.currentTimeMillis();

			if (status == Status.RUNNING) {
				if (wrapper.getExpireTime() < now) {
					status = Status.CANCELLING;
					if (thread != null)
						thread.interrupt();
					cancelTime = System.currentTimeMillis();
				}
			} else if (status == Status.CANCELLING) {
				if (cancelTime + KILL_TIME < now) {
					status = Status.DEAD;
				}
			}

		} finally {
			lock.unlock();
		}

	}

	@SuppressWarnings("deprecation")
	public void kill() {

		// System.err.println("kill...");

		if (status != Status.DEAD)
			return;

		if (wrapper != null)
			wrapper.done(new InterruptedException("timeout, thread is killed"));

		wrapper = null;

		if (thread != null) {
			while (!isDead())
				thread.stop();
		}
	}

	public boolean isDead() {
		return thread == null || !thread.isAlive();
	}

}
